package com.heima.kafka.sample;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @author 谢金成
 * @date 2024/4/12 11:38
 */
public class ProducerQuickStart {
    public static void main(String[] args) {
        // 1.kafka链接配置信息
        Properties properties = new Properties();
        // kafka链接地址
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "10.0.0.2:9092");
        // key和value的序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");

        // acks 消息确认机制
        // 0 认为不会发生错误, 发送消息后立即返回, 消息可能丢失
        // 1 只要集群的leader节点收到消息, 生产者就会收到成功响应 消息可能丢失
        // all 只要所有参与的节点全部收到消息, 生产者才会收到响应 消息可能重复消费

        properties.put(ProducerConfig.ACKS_CONFIG, "all");

        // retries, 重试次数
        properties.put(ProducerConfig.RETRIES_CONFIG, 10);

        // 消息压缩
        // snappy 占用较少的cpu资源, 有可观的压缩比
        // lz4 占用较少的cpu, 压缩和解压缩速度快, 压缩比可观
        // gzip 占用较多的cpu, 会提高压缩比
        properties.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, "snappy");

        // 2.创建kafka生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        // 3.发送消息
        /**
         * topic 主题
         * partition 分区
         * key
         * value
         */

        for (int i = 0; i < 5; i++) {
            ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>("itcast-topic-input", "hello Kafka");
            producer.send(producerRecord);
        }
        // 同步发送
//        try {
//            recordMetadata = producer.send(producerRecord).get();
//            System.out.println(recordMetadata.offset());
//        } catch (InterruptedException e) {
//            e.printStackTrace();
//        } catch (ExecutionException e) {
//            e.printStackTrace();
//        }
//         // 异步发送
//         producer.send(producerRecord, new Callback() {
//             @Override
//             public void onCompletion(RecordMetadata recordMetadata, Exception e) {
//                 if (e != null) {
//                     System.out.println("记录异常信息到日志表中");
//                 }
//
//                 System.out.println(recordMetadata.offset());
//             }
//         });
//
        // 4.关闭信息通道
        producer.close();
    }
}
